//! Redis链路驱动层

use futures_util::StreamExt;
use redis::{aio, AsyncCommands, Client};
use unmp_link::prelude::*;

pub struct Conn {
    identifier: String,
    topic: String,
    pub_ch: aio::Connection,
    sub_ch: aio::PubSub,
}
impl Conn {
    /// 创建一个UDP链路实例，并连接到指定地址端口
    pub async fn new(url: &str, topic: String) -> Result<Connection, ()> {
        let client = Client::open(url).map_err(|_| ())?;
        let identifier = format!("{}/{}", client.get_connection_info().addr, topic);
        let pub_ch = client.get_async_connection().await.map_err(|_| ())?;
        let sub_ch = client.get_async_connection().await.map_err(|_| ())?;
        let mut sub_ch = sub_ch.into_pubsub();
        sub_ch.subscribe(&topic).await.map_err(|_| ())?;

        Ok(Connection::new(Box::new(Self {
            identifier,
            topic,
            pub_ch,
            sub_ch,
        })))
    }
}
#[async_trait]
impl ConnectionDriver for Conn {
    fn is_subnet(&self) -> bool {
        false
    }
    fn kind(&self) -> &str {
        "redis"
    }
    fn id(&self) -> &str {
        &self.identifier
    }
    async fn send(&mut self, buf: Bytes) -> Result<(), SendError> {
        if let Err(_) = self
            .pub_ch
            .publish::<_, _, u8>(&self.topic, buf.as_ref())
            .await
        {
            return Err(SendError::Disconnected);
        }
        Ok(())
    }
    async fn recv(&mut self) -> Result<Bytes, RecvError> {
        let msg = if let Some(msg) = self.sub_ch.on_message().next().await {
            msg
        } else {
            return Err(RecvError::Disconnected);
        };
        let buf = msg.get_payload_bytes();
        Ok(Bytes::copy_from_slice(buf))
    }
}
